package cn.itcast.flink.stream;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 *  * DataStream数据流与Table表之间相互转换，在实际项目中，
 *  可以先基于DataStream记进行数据过滤转换操作，然后转换为Table，最后使用SLQ分析
 * @author lilulu
 */
public class StreamToTableDemo {
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class OrderInfo {
        private String userId;
        private Long ts;
        private Double money;
        private String category;
    }
    public static void main(String[] args) throws Exception {
        // 1. 执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //创建流式表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, EnvironmentSettings.newInstance().inStreamingMode().useBlinkPlanner().build());
        // 2. 数据源-source
        DataStreamSource<String> dataStreamSource = env.readTextFile("datas/order.csv");
        // 3. 数据转换-transformation
        SingleOutputStreamOperator<OrderInfo> orderInfo = dataStreamSource.map(new MapFunction<String, OrderInfo>() {
            @Override
            public OrderInfo map(String line) throws Exception {
                String[] split = line.split(",");
                OrderInfo orderInfo = new OrderInfo();
                orderInfo.setUserId(split[0]);
                orderInfo.setTs(Long.parseLong(split[1]));
                orderInfo.setMoney(Double.parseDouble(split[2]));
                orderInfo.setCategory(split[3]);
                return orderInfo;
            }
        });
        Table orderTable = tableEnv.fromDataStream(orderInfo);
        tableEnv.createTemporaryView("tbl_order",orderTable);

        Table resultTable = tableEnv.sqlQuery("select * from tbl_order");
        DataStream<Row> rowDataStream = tableEnv.toDataStream(resultTable);
        rowDataStream.printToErr();
        // 4. 数据终端-sink
        // 5. 触发执行-execute
        env.execute("StreamToTableDemo");
    }
}